/**
 * This file is a part of qloudgen-rul. 
 * You can redistribute qloudgen-rul and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-rul is distributed WITHOUT ANY WARRANTY. 
 *
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/. 
 **/
package org.qloudgen.rul.curator;

import java.io.IOException;

import java.util.Map;
import java.util.HashMap;
import java.util.Properties;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.curator.RetryPolicy;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.retry.RetryOneTime;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.BoundedExponentialBackoffRetry;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.CuratorFrameworkFactory.Builder;
import org.apache.curator.utils.CloseableUtils;

import org.qloudgen.sec.util.StringUtil;
import org.qloudgen.sec.util.ResourceUtil;
import org.qloudgen.sec.util.ExceptionUtil;

/**
 * <p>Title: Curator Framework Wrapper</p>
 * <p>Description: Wrapper curator framework implementation.</p>
 * <p>Copyright: qloudgen/rul 2014</p>
 * @author <a href="mailto:hyq.dd.sun@gmail.com">glue.3M</a>
 */
public abstract class AbstractCuratorClient implements CuratorClient {
	/**
	 * Log output.
	 */
	private final static Logger logger = LoggerFactory.getLogger( AbstractCuratorClient.class );
	/**
	 * Default curator config file path.
	 */
	private final static String DEFAULT_CURATOR_FILE = "classpath:curator.properties";
	/**
	 * Current curator configuration.
	 */
	private Map<String,Object> config = null;
	/**
	 * Current curator framework.
	 */
	private CuratorFramework client = null;
	
	/**
	 * Default constructor.
	 */
	public AbstractCuratorClient() {
		this( System.getProperty( "qloudgen.curator" , null ) );
	}
	
	/**
	 * Custom constructor.
	 *
	 * @param configFile
	 */
	public AbstractCuratorClient(String configFile) {
		//
		try {
			String file = null;
			if ( StringUtil.isBlank( configFile ) ) {
				logger.info( "Using default curator config file...{}" , DEFAULT_CURATOR_FILE );
				//
				file = DEFAULT_CURATOR_FILE;
			}
			else {
				logger.info( "Using custom curator config file...{}" , configFile );
				//
				file = configFile;
			}
			//
			config = getClientConfig( file );
		}
		catch (Throwable t) {
			logger.error( "Current curator framework configuration error...\n{}" , ExceptionUtil.get( t ) );
			throw new CuratorException( t );
		}
	}
	
	/**
	 * Initailize.
	 */
	public void init() {
		logger.info( "Starting initialize curator framework..." );
		//
		Builder b = CuratorFrameworkFactory
		                .builder()
		                .connectString( ( String )config.get( "connectString" ) )
		                .retryPolicy( ( RetryPolicy )config.get( "retryPolicy" ) );
		//
		if ( config.containsKey( "sessionTimeoutMs" ) ) {
			b = b.sessionTimeoutMs( Integer.parseInt( ( String )config.get( "sessionTimeoutMs" ) ) );
		}
		if ( config.containsKey( "connectionTimeoutMs" ) ) {
			b = b.connectionTimeoutMs( Integer.parseInt( ( String )config.get( "connectionTimeoutMs" ) ) );
		}
		/**
		 * Create & start.
		 */
		client = b.build();
		client.start();
		//
		logger.info( "Started initialize curator framework...{}" , this );
	}
	
	/**
	 * Destroy.
	 */
	public void destroy() {
		logger.info( "Starting destroy curator framework..." );
		CloseableUtils.closeQuietly( client );
		client = null;
		config = null;
		logger.info( "Started destroy curator framework..." );
	}
	
	/**
	 * Returns current curator framework.
	 *
	 * @return CuratorFramework
	 */
	protected CuratorFramework getClient() {
		return client;
	}
	
	/**
	 * Returns current curator config.
	 *
	 * @param file
	 * @return Map<String,Object>
	 */
	private Map<String,Object> getClientConfig(String file) throws IOException {
		//
		Properties props = new Properties();
		props.load( ResourceUtil.getInputStream( file ) );
		//
		logger.info( "Current curator configuration...{}" , props );
		/**
		 * Do parse global parameter.
		 */
		Map<String,Object> current = new HashMap<String,Object>();
		current.put( "connectString" , props.getProperty( "zk.connectString" ) );
		if ( ! StringUtil.isBlank( props.getProperty( "zk.sessionTimeoutMs" ) ) ) {
			current.put( "sessionTimeoutMs" , props.getProperty( "zk.sessionTimeoutMs" ) );
		}
		if ( ! StringUtil.isBlank( props.getProperty( "zk.connectionTimeoutMs" ) ) ) {
			current.put( "connectionTimeoutMs" , props.getProperty( "zk.connectionTimeoutMs" ) );
		}
		/**
		 * Do parse retry policy.
		 */
		String policy = props.getProperty( "zk.retryPolicy" );
		int maxRetries = ( StringUtil.isBlank( props.getProperty( "zk.retryPolicy.maxRetries" ) ) ? -1 : Integer.parseInt( props.getProperty( "zk.retryPolicy.maxRetries" ) ) );
		int baseSleepTimeMs = ( StringUtil.isBlank( props.getProperty( "zk.retryPolicy.baseSleepTimeMs" ) ) ? -1 : Integer.parseInt( props.getProperty( "zk.retryPolicy.baseSleepTimeMs" ) ) );
		int maxSleepMs = ( StringUtil.isBlank( props.getProperty( "zk.retryPolicy.maxSleepMs" ) ) ? -1 : Integer.parseInt( props.getProperty( "zk.retryPolicy.maxSleepMs" ) ) );
		int maxSleepTimeMs = ( StringUtil.isBlank( props.getProperty( "zk.retryPolicy.maxSleepTimeMs" ) ) ? -1 : Integer.parseInt( props.getProperty( "zk.retryPolicy.maxSleepTimeMs" ) ) );
		int sleepMsBetweenRetry = ( StringUtil.isBlank( props.getProperty( "zk.retryPolicy.sleepMsBetweenRetry" ) ) ? -1 : Integer.parseInt( props.getProperty( "zk.retryPolicy.sleepMsBetweenRetry" ) ) );
		int sleepMsBetweenRetries = ( StringUtil.isBlank( props.getProperty( "zk.retryPolicy.sleepMsBetweenRetries" ) ) ? -1 : Integer.parseInt( props.getProperty( "zk.retryPolicy.sleepMsBetweenRetries" ) ) );
		int maxElapsedTimeMs = ( StringUtil.isBlank( props.getProperty( "zk.retryPolicy.maxElapsedTimeMs" ) ) ? -1 : Integer.parseInt( props.getProperty( "zk.retryPolicy.maxElapsedTimeMs" ) ) );
		//
		RetryPolicy retryPolicy = null;
		if ( policy.equals( "ExponentialBackoffRetry" ) ) {
			if ( maxSleepMs == -1 ) {
				retryPolicy = new ExponentialBackoffRetry( baseSleepTimeMs , maxRetries );
			}
			else {
				retryPolicy = new ExponentialBackoffRetry( baseSleepTimeMs , maxRetries , maxSleepMs );
			}
		}
		else if ( policy.equals( "BoundedExponentialBackoffRetry" ) ) {
			retryPolicy = new BoundedExponentialBackoffRetry( baseSleepTimeMs , maxSleepTimeMs , maxRetries );
		}
		else if ( policy.equals( "RetryOneTime" ) ) {
			retryPolicy = new RetryOneTime( sleepMsBetweenRetry );
		}
		else if ( policy.equals( "RetryNTimes" ) ) {
			retryPolicy = new RetryNTimes( maxRetries , sleepMsBetweenRetries );
		}
		else if ( policy.equals( "RetryUntilElapsed" ) ) {
			retryPolicy = new RetryUntilElapsed( maxElapsedTimeMs , sleepMsBetweenRetries );
		}
		else {
			throw new IOException( "Unknown retry policy implementation..." );
		}
		current.put( "retryPolicy" , retryPolicy );
		//
		return current;
	}
	
	/**
	 * Returns the String representation.
	 *
	 * @return String
	 */
	@Override
	public String toString() {
		StringBuilder builder = new StringBuilder();
		builder.append( "[" );
		builder.append( super.toString() );
		builder.append( "-->" );
		builder.append( "client=" ).append( client ).append( "," );
		builder.append( "config=" ).append( config );
		builder.append( "]" );
		//
		return builder.toString();
	}
}